package com.xxx.rpc.registry.zookeeper;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.collections4.map.MultiValueMap;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.jboss.netty.util.internal.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.xxx.rpc.common.bean.IpPort;
import com.xxx.rpc.common.util.StringUtil;

/**
 * 服务的观察者  缓存 ，动态服务管理中心；
 * @author Rosun
 *
 */
public class ZookeeperServiceCenter implements ServiceCenter{
	private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperServiceCenter.class);
	private CuratorFramework client = null;
	private CountDownLatch countDownLatch = new CountDownLatch(1);
	private String zkAddress = "127.0.0.1:2181";
	private TreeCache treeCache;

	/**
	 * key: serviceNameWithVersion,接口名称（带版本）;<br>
	 * value: 提供该服务的机器列表；
	 */
	private MultiValueMap<String, PathDataPair> nodeDataMap = MapUtils
			.multiValueMap(new ConcurrentHashMap<String, Collection<PathDataPair>>());

	public ZookeeperServiceCenter(String zkAddress) {
		if (StringUtil.isNotEmpty(zkAddress)) {
			this.zkAddress = zkAddress;
		}
	}

//	private List<IpPort> list = null;
	/**
	 * 获取服务提供者信息；
	 * @param serviceNameWithVersion 接口名称（带版本）;
	 * @return 提供该服务的机器列表；
	 */
	public List<IpPort> getServerStub(String serviceNameWithVersion){
		Collection<PathDataPair> pps =  this.nodeDataMap.getCollection(serviceNameWithVersion);
		if(pps == null || pps.size() == 0){
			return null;
		}
		
//		if(this.list == null){
//			list = new ArrayList<>();
//			for(PathDataPair pp : pps){
//				String data = pp.getData();
//				if(StringUtil.isNotEmpty(data)){
//					String[] split = data.split(":");
//					IpPort ipport = new IpPort();
//					ipport.setIp(split[0]);
//					ipport.setPort(Integer.parseInt(split[1]));
//					list.add(ipport);
//				}
//			}
//		}
		List<IpPort> list = new ArrayList<>();
		for(PathDataPair pp : pps){
			String data = pp.getData();
			if(StringUtil.isNotEmpty(data)){
				String[] split = data.split(":");
				IpPort ipport = new IpPort();
				ipport.setIp(split[0]);
				ipport.setPort(Integer.parseInt(split[1]));
				list.add(ipport);
			}
		}
		return list;
	}
	
	/**
	 * 服务监控客户端初始化
	 * @throws Exception
	 */
	/* (non-Javadoc)
	 * @see com.xxx.rpc.registry.zookeeper.ServiceCenter#createWatchClient()
	 */
	public void createWatchClient() throws Exception {
		LOGGER.info("createWatchClient : " + this.zkAddress);
		client = CuratorFrameworkFactory.builder().connectString(zkAddress)
				.retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
				.connectionTimeoutMs(Constant.ZK_CONNECTION_TIMEOUT).sessionTimeoutMs(Constant.ZK_SESSION_TIMEOUT)
				.build();
		// 客户端注册监听，进行连接配置
		client.getConnectionStateListenable().addListener(connectionListener);

		client.start();
		// 连接成功后，才进行下一步的操作
		countDownLatch.await();
		this.watchTreePath();
		//等待一秒，加载zookeeper上的数据。
		Thread.sleep(1000);
	}

	private void closeClient() {
		if(treeCache != null){
			treeCache.close();
		}
		if (this.client != null && this.client.getState() != CuratorFrameworkState.STOPPED) {
			this.client.close();
			this.client = null;
		}
	}

	/**
	 * 客户端重新注册
	 */
	private void reinit() {
		LOGGER.info("need reinit....");
		try {
			createWatchClient();
		} catch (Exception e) {
			LOGGER.error("failed reinit," + e.getMessage());
			this.client = null;
		}
	}

	public static void main(String[] args) throws Exception {
		
//		ZookeeperServiceCenter pp = new ZookeeperServiceCenter("192.168.2.16:2181");
		ZookeeperServiceCenter pp = new ZookeeperServiceCenter("127.0.0.1:2181");
		pp.createWatchClient();
		LOGGER.debug("init ok........");
		Thread.sleep(Integer.MAX_VALUE);
	}

	// private String directPath ;
	// private String treePath ;
	// private PathChildrenCache pathCache ;
	

	// // 对path进行监听配置
	// public void watchDirectPath(String directPath) throws Exception {
	// this.directPath = directPath;
	// // 子节点的监听
	// this.pathCache = new PathChildrenCache(client, directPath, false);
	// pathCache.start();
	// // 注册监听
	// pathCache.getListenable().addListener(plis);
	//
	//
	// }

	private void debugNodeDatasMap() {
		LOGGER.debug(nodeDataMap.toString());
	}

	/**
	 * 对path进行监听配置
	 * @param treePath
	 * @throws Exception
	 */
	private void watchTreePath( ) throws Exception {
		LOGGER.info("pwatchTreePath (" + Constant.ZK_REGISTRY_PATH + ") ... ");
		// 对path路径下所有孩子节点的监听
		this.treeCache = new TreeCache(client, Constant.ZK_REGISTRY_PATH);
		treeCache.start();
		treeCache.getListenable().addListener(treeCacheListener);
	}

	// 客户端的监听配置
	private ConnectionStateListener connectionListener = new ConnectionStateListener() {
		public void stateChanged(CuratorFramework client, ConnectionState newState) {
			System.err.println("########## stateChanged ########## :newState=" + newState);
			if (newState == ConnectionState.CONNECTED) {
				LOGGER.debug("connected established ");
				countDownLatch.countDown();
			} else if (newState == ConnectionState.RECONNECTED) {
				LOGGER.debug("connected RECONNECTED RECONNECTED");
				// 需要重新读取下 zookeeper 节点数据并本地缓存起来；??
				debugNodeDatasMap();
			} else if (newState == ConnectionState.SUSPENDED) {
				LOGGER.debug("connected SUSPENDED SUSPENDED");
				// 需要重新读取下 zookeeper 节点数据并本地缓存起来；??
				debugNodeDatasMap();
			} else if (newState == ConnectionState.LOST) {
				LOGGER.warn("connection lost,waiting for reconection");
				try {
					ZookeeperServiceCenter.this.closeClient();
					while (ZookeeperServiceCenter.this.client == null) {
						LOGGER.debug("-- reinit---");
						reinit();
						Thread.sleep(1000);
						LOGGER.debug("--  reinit end---");
					}
				} catch (Exception e) {
					LOGGER.error("re-inited failed",e);
				}
			}

		}
	};

	/**
	 * 子节点的监听
	 */
	// private PathChildrenCacheListener plis = new PathChildrenCacheListener()
	// {
	//
	// public void childEvent(CuratorFramework client, PathChildrenCacheEvent
	// event) throws Exception {
	// switch (event.getType()) {
	// case CHILD_ADDED: {
	// System.out.println("PathChildrenCacheListener Node added: " +
	// ZKPaths.getNodeFromPath(event.getData().getPath()) + " , data:"
	// + ss.deserialize(event.getData().getData()) );
	// break;
	// }
	//
	// case CHILD_UPDATED: {
	// System.out.println("PathChildrenCacheListener Node changed: " +
	// ZKPaths.getNodeFromPath(event.getData().getPath()) + " , data: "
	// + ss.deserialize(event.getData().getData()) );
	// break;
	// }
	//
	// case CHILD_REMOVED: {
	// System.out.println("PathChildrenCacheListener Node removed: " +
	// ZKPaths.getNodeFromPath(event.getData().getPath()) );
	// break;
	// }
	// default:
	// System.out.println("PathChildrenCacheListener default...." +
	// ZKPaths.getNodeFromPath(event.getData().getPath()) );
	// break;
	// }
	//
	// }
	// };

	private SerializableSerializer ss = new SerializableSerializer();
	/**
	 * 所有子节点的监听
	 */
	private TreeCacheListener treeCacheListener = new TreeCacheListener() {

		public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
			if(event.getData() == null || event.getData().getPath() == null){
				//回避 zookeeper 的空指针异常问题；by 罗珊；
				return ;
			}
			String[] split = event.getData().getPath().split("/");
			if (split.length < 2) {
				System.err.println("IGNORED PATH ,too short for processing :" + event.getData().getPath());
				return;
			}
			String serviceNameWithVersion = split[split.length - 2];
			LOGGER.debug("childEvent for : " + serviceNameWithVersion);
			if (Constant.ZK_REGISTRY_PATH.equals(event.getData().getPath())
					|| Constant.ZK_REGISTRY_PATH.contains(event.getData().getPath())) {
				System.err.println("ignore base path ：" + event.getData().getPath());
				return;
			}
			switch (event.getType()) {
			case NODE_ADDED:
				if(event.getData().getData() == null || event.getData().getData().length ==0){
					//System.err.println("ignore path add event：" + event.getData().getPath());
					return ;
				}
				String ipPortData = (String) ss.deserialize(event.getData().getData());
				LOGGER.debug(
						"TreeCacheListener TreeNode added: " + event.getData().getPath() + " , data: " + ipPortData);
				PathDataPair pdp = new PathDataPair(event.getData().getPath(), ipPortData);
				nodeDataMap.put(serviceNameWithVersion, pdp);
				break;
			case NODE_UPDATED:
				System.err.println("TreeCacheListener TreeNode updated: " + event.getData().getPath() + " , data: "
						+ ss.deserialize(event.getData().getData()));
				break;
			case NODE_REMOVED:
				PathDataPair pdp2 = new PathDataPair(event.getData().getPath());
				LOGGER.debug("TreeCacheListener TreeNode removed: " + event.getData().getPath());
				if (nodeDataMap.getCollection(serviceNameWithVersion).contains(pdp2)) {
					nodeDataMap.getCollection(serviceNameWithVersion).remove(pdp2);
				}
				break;
			default:
				LOGGER.debug("TreeCacheListener TreeNode : " + event.getType());
				break;
			}
			LOGGER.debug("\n\n");
			debugNodeDatasMap();
			LOGGER.debug("\n\n");
		}
	};
}
